package netty.server;


import common.Constants;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.util.concurrent.ExecutionException;


public class StringHandler extends SimpleChannelInboundHandler<String> {
    public static final Logger LOG = LoggerFactory.getLogger(StringHandler.class);
    //private long a=0;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        //get the message
        LOG.debug(ctx.channel().remoteAddress() + " Say : " + msg);
        //invoke kafka api
        //check it was heard beat test
        if(msg !=null && msg.equalsIgnoreCase(Constants.CLIENT_HEARD_BEAT_TEST_CONTENT)){
            LOG.info("client "+ctx.channel().remoteAddress() + " send a heard beat test : " + msg);
            ctx.writeAndFlush(Constants.RESPONSE_SUCCESS);
            return;
        }
        try {
            Server.kafkaClient.sendMessage(Server.kafkaMessageKey, msg);
        } catch (ExecutionException e) {
            LOG.error("send msg to kafka server is failed.");
            e.printStackTrace();
            return;
        } catch (InterruptedException e) {
            LOG.error("send msg to kafka server is failed.");
            e.printStackTrace();
            return;
        }
        //a++;
        //LOG.info("client had send :"+a);
        //response to client "we received it"
        ctx.writeAndFlush(Constants.RESPONSE_SUCCESS);
    }

    /*
     *
     *when client connect to server
     * we print client address
     * we response welcome to xx service
     * */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        LOG.info("RamoteAddress : " + ctx.channel().remoteAddress() + " active !");
        //ctx.writeAndFlush("Welcome to " + InetAddress.getLocalHost().getHostName() + " service!\n");
        super.channelActive(ctx);
    }
}